package edu.nepu.flink.api.window;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Date 2024/2/29 21:49
 * @Created by chenshuaijun
 */
public class WindowLateTime {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
            }
        });
        // 提取事件时间,指定watermark的延迟时间
        SingleOutputStreamOperator<WaterSensor> waterSource = source.assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }));

        /**
         * 我们需要着重聊一下这个窗口允许的迟到时间
         * 它指的不是延迟窗口的触发时间，而是延迟窗口的关闭时间
         * 当事件时间到达触发窗口计算的时间的时候窗口会正常的计算，
         * 但是窗口不会关闭，只要事件时间 - 窗口允许迟到时间 < 窗口关闭时间。那么属于本窗口的数据到来的时候，就会触发再次的计算，并且是每来一条就计算一次
         */
        // 将迟到的数据放入到侧输出流中
        OutputTag<WaterSensor> lateStream = new OutputTag<WaterSensor>("later-data"){};
        SingleOutputStreamOperator<WaterSensor> reduce = waterSource
                .keyBy(WaterSensor::getId).
                window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(2)) // 设置允许迟到的时间是2s
                .sideOutputLateData(lateStream)
                .reduce((v1, v2) -> new WaterSensor(v1.id, v1.ts, v1.vc + v2.vc));

        reduce.print("主流的数据");
        SideOutputDataStream<WaterSensor> sideOutput = reduce.getSideOutput(lateStream);
        sideOutput.printToErr("侧输出流的数据");


        env.execute();
    }
}
